package com.cnpc.bds.sdk.sink.kafka.meta;

import com.cnpc.bds.sdk.sink.kafka.support.unit.DataSize;
import com.cnpc.bds.sdk.sink.kafka.support.unit.DataUnit;
import lombok.Builder;
import lombok.Data;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;

import java.util.Properties;

/**
 * Kafka生产者配置.
 */
@Data
@Builder
public class ProducerCfg {

    private TopicMeta topicMeta;
    /**
     * config.
     */
    private String keySerializer;
    private String valueSerializer;
    private DataSize batchSize;
    private DataSize bufferMemory;
    private Integer lingerMs;
    private Integer maxBlockMs;

    public Properties properties() {
        Properties config = new Properties();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, topicMeta.getBrokers());
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, (int) batchSize.toBytes());
        config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, (int) bufferMemory.toBytes());
        config.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
        config.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
        return config;
    }

    public static ProducerCfgBuilder defaultCfg() {
        return ProducerCfg.builder()
                .keySerializer(ByteArraySerializer.class.getName())
                .valueSerializer(ByteArraySerializer.class.getName())
                .batchSize(DataSize.of(4, DataUnit.KILOBYTES))
                .bufferMemory(DataSize.of(32, DataUnit.MEGABYTES))
                .lingerMs(250)
                .maxBlockMs(2000);
    }


}
